#include "FilterOperationNode.h"
#include "ExpressionNode.h"
#include "config.h"

FilterOperationNode::FilterOperationNode(OperationNode* source, ExpressionNode* expression):
		_source(source),
		_expression(expression),
		_remainder(),
		_current(0) {}

FilterOperationNode::~FilterOperationNode() {
	delete _source;
	delete _expression;
}

template <typename T>
void filter(T* out, T* remainder, const T* in, const bool* condition, int n, int& cur) {
	T* buffers[2] = {out, remainder};

	for (int i = 0; i < n; ++i) {
		buffers[cur / PULL_AMOUNT][cur % PULL_AMOUNT] = in[i];
		cur += condition[i];
	}
}

SharedArray<Column> FilterOperationNode::pull() {
	SharedArray<Column> in = _source->pull();
	assert(in.size() > 0);

	SharedArray<Column> out(in.size());
	for (int col = 0; col < in.size(); ++col) {
		out[col] = Column(in[col].type(), PULL_AMOUNT);
	}

	if (_remainder.size() == 0) {
		_remainder = SharedArray<Column>(in.size());
		for (int col = 0; col < in.size(); ++col) {
			_remainder[col] = Column(in[col].type(), PULL_AMOUNT);
		}
	} else {
		assert(_remainder.size() == in.size());
		out.swap(_remainder);
	}

	while (true) {
		int colSize = in[0].size();
		assert(colSize <= PULL_AMOUNT);

		if (colSize == 0) {
			for (int col = 0; col < out.size(); ++col) {
				out[col].shrink(_current);
			}
			_current = 0;
			break;
		}

		Column condition = _expression->pull(in);
		assert(condition.type() == BOOL);
		int currentBefore = _current;

		for (int col = 0; col < in.size(); ++col) {
			assert(in[col].size() == colSize);
			assert(in[col].type() == out[col].type());
			assert(in[col].type() == _remainder[col].type());
			_current = currentBefore;

			switch (in[col].type()) {
				case INT:
					filter(out[col].data<int32_t>(), _remainder[col].data<int32_t>(),
							in[col].data<int32_t>(), condition.data<bool>(),
							in[col].size(), _current);
					break;

				case DOUBLE:
					filter(out[col].data<double>(), _remainder[col].data<double>(),
							in[col].data<double>(), condition.data<bool>(),
							in[col].size(), _current);
					break;

				case BOOL:
					filter(out[col].data<bool>(), _remainder[col].data<bool>(),
							in[col].data<bool>(), condition.data<bool>(),
							in[col].size(), _current);
					break;

				default:
					assert(false);
			}
		}

		if (_current < PULL_AMOUNT) {
			in = _source->pull();
		} else {
			_current %= PULL_AMOUNT;
			break;
		}
	}
	
	return out;
}
